 1.Kafka高级特性解析之稳定性中 __consumer_offsets
   
   Zookeeper不适合大批量的频繁写入操作。
   Kafka 1.0.2将consumer的位移信息保存在Kafka内部的topic中，即__consumer_offsets主题，并且默认提供了
kafka_consumer_groups.sh脚本供用户查看consumer信息。
   1).创建topic “tp_test_01”
   [root@linux121 ~]# kafka-topics.sh --zookeeper linux121:2181/myKafka --create --topic
tp_test_01 --partitions 5 --replication-factor 1
   2).使用kafka-console-producer.sh脚本生产消息
   [root@linux121 ~]# for i in `seq 100`; do echo "hello lagou $i" >> messages.txt; done
   [root@linux121 ~]# kafka-console-producer.sh --broker-list linux121:9092 --topic tp_test_01 <
messages.txt
   由于默认没有指定key，所以根据round-robin方式，消息分布到不同的分区上。 (本例中生产了100条消息)
   3).验证消息生成成功
   [root@linux121 ~]# kafka-console-producer.sh --broker-list linux121:9092 --topic tp_test_01
< messages.txt
>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>[root@linux121 ~]# kafka-run-class.sh kafka.tools.GetOffsetShell --broker-list
linux121:9092 --topic tp_test_01 --time -1
tp_test_01:2:20
tp_test_01:4:20
tp_test_01:1:20
tp_test_01:3:20
tp_test_01:0:20
   [root@linux121 ~]#
   结果输出表明100条消息全部生产成功！
   4).创建一个console consumer group
   [root@linux121 ~]#kafka-console-consumer.sh --bootstrap-server linux121:9092 --topic
tp_test_01 --from-beginning
   5).获取该consumer group的group id(后面需要根据该id查询它的位移信息)
   [root@linux121 ~]# kafka-consumer-groups.sh --bootstrap-server linux121:9092 --list
   输出： console-consumer-84373 (记住这个id！)
   [root@linux121 ~]# kafka-consumer-groups.sh --bootstrap-server linux121:9092 --describe
--group console-consumer-84373
   6).查询__consumer_offsets topic所有内容
   注意：运行下面命令前先要在consumer.properties中设置exclude.internal.topics=false
   [root@linux121 ~]# kafka-console-consumer.sh --topic __consumer_offsets --bootstrap-server
linux121:9092 --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter" 
--consumer.config /opt/kafka_2.12-1.0.2/config/consumer.properties --from-beginning
   默认情况下__consumer_offsets有50个分区，如果你的系统中consumer group也很多的话，那么这个命令的输
出结果会很多。
   7).计算指定consumer group在__consumer_offsets topic中分区信息
   这时候就用到了第5步获取的group.id(本例中是console-consumer-84373)。Kafka会使用下面公式计算该group
位移保存在__consumer_offsets的哪个分区上：
package com.lagou.kafka.test;

public class MyTest {
    public static void main(String[] args) {
        String str = "console-consumer-84373";
        // _comsumer_offsets的主题分区数量为50个
        System.out.println(Math.abs(str.hashCode()) % 50);
    }
}
   对应的分区=Math.abs("console-consumer-84373".hashCode()) % 50 = 26，即__consumer_offsets的分区19保
存了这个consumer group的位移信息。
   8).获取指定consumer group的位移信息
   [root@linux121 ~]# kafka-simple-consumer-shell.sh --topic __consumer_offsets --partition 26 --
broker-list linux121:9092 --formatter "kafka.coordinator.group.GroupMetadataManager\$OffsetsMessageFormatter"
   下面是输出结果：
...
[console-consumer-84373,tp_test_01,3]::[OffsetMetadata[20,NO_METADATA],CommitTime
1596424702212,ExpirationTime 1596511102212]
[console-consumer-84373,tp_test_01,4]::[OffsetMetadata[20,NO_METADATA],CommitTime
1596424702212,ExpirationTime 1596511102212]
[console-consumer-84373,tp_test_01,0]::[OffsetMetadata[20,NO_METADATA],CommitTime
1596424702212,ExpirationTime 1596511102212]
[console-consumer-84373,tp_test_01,1]::[OffsetMetadata[20,NO_METADATA],CommitTime
1596424702212,ExpirationTime 1596511102212]
[console-consumer-84373,tp_test_01,2]::[OffsetMetadata[20,NO_METADATA],CommitTime
1596424702212,ExpirationTime 1596511102212]
[console-consumer-84373,tp_test_01,3]::[OffsetMetadata[20,NO_METADATA],CommitTime
1596424707212,ExpirationTime 1596511107212]
[console-consumer-84373,tp_test_01,4]::[OffsetMetadata[20,NO_METADATA],CommitTime
1596424707212,ExpirationTime 1596511107212]
[console-consumer-84373,tp_test_01,0]::[OffsetMetadata[20,NO_METADATA],CommitTime
1596424707212,ExpirationTime 1596511107212]
...
   上图可见，该consumer group果然保存在分区11上，且位移信息都是对的(这里的位移信息是已消费的位移，严
格来说不是第3步中的位移。由于我的consumer已经消费完了所有的消息，所以这里的位移与第3步中的位移相同)。
另外，可以看到__consumer_offsets topic的每一日志项的格式都是：[Group, Topic, Partition]::
[OffsetMetadata[Offset, Metadata], CommitTime, ExpirationTime]。
